//服务注册
package registry

import (
	"bytes"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"sync"
	"time"
)

const Serverport = ":3000"
const ServicesURL = "http://localhost" + Serverport + "/services"

//注册中心结构体
type registry struct {
	registrations []Registration //维护的注册服务的列表
	mutex         *sync.RWMutex
}

//注册添加服务函数
func (r *registry) add(reg Registration) error {
	r.mutex.Lock()
	r.registrations = append(r.registrations, reg)
	r.mutex.Unlock()
	//将已经在服务中心注册的服务的依赖 告诉服务
	if err := r.sendExistRequiredServices(reg); err != nil {
		return err
	}
	//通知已有服务本服务已经上线
	r.notify(patch{Added: []patchEntry{
		{Name: reg.ServiceName,
			URL: reg.ServiceURL},
	}})
	return nil
}

//注册删除服务函数
func (r *registry) remove(url string) error {
	for i := range reg.registrations {
		if reg.registrations[i].ServiceURL == url {
			//通知删除
			r.notify(patch{
				Removed: []patchEntry{
					{Name: reg.registrations[i].ServiceName,
						URL: reg.registrations[i].ServiceURL},
				},
			})
			r.mutex.Lock()
			//真正删除
			reg.registrations = append(reg.registrations[:i], reg.registrations[i+1:]...)
			r.mutex.Unlock()
			return nil
		}
	}
	return fmt.Errorf("Service at URL %s not found", url)
}

//用patch去更新
func (r registry) notify(fullPatch patch) {
	r.mutex.RLock()
	defer r.mutex.RUnlock()
	//对于目前已有的服务
	for _, reg := range r.registrations {
		go func(reg Registration) {
			//遍历每个服务的依赖服务
			for _, reqService := range reg.RequiredServices {
				//更新的patch
				p := patch{Added: []patchEntry{}, Removed: []patchEntry{}}
				needUpdate := false
				//如果列表内新增的有这个依赖
				for _, added := range fullPatch.Added {
					if added.Name == reqService {
						p.Added = append(p.Added, added)
						needUpdate = true
					}
				}
				for _, removed := range fullPatch.Removed {
					if removed.Name == reqService {
						p.Removed = append(p.Removed, removed)
						needUpdate = true
					}
				}
				if needUpdate {
					if err := r.sendPatch(p, reg.ServiceUpdateURL); err != nil {
						log.Println(err)
						return
					}
				}
			}
		}(reg)
	}
}

//将已经在服务中心注册的服务的依赖 告诉服务
func (r registry) sendExistRequiredServices(reg Registration) error {
	r.mutex.RLock()
	defer r.mutex.RUnlock()
	//确定已注册的依赖
	var p patch
	for _, serviceReg := range r.registrations {
		for _, regService := range reg.RequiredServices {
			if serviceReg.ServiceName == regService {
				p.Added = append(p.Added,
					patchEntry{
						Name: serviceReg.ServiceName,
						URL:  serviceReg.ServiceURL,
					})
			}
		}
	}
	//向服务指定的更新URL发送更新列表
	if err := r.sendPatch(p, reg.ServiceUpdateURL); err != nil {
		return err
	}
	return nil
}

//
func (r registry) sendPatch(p patch, url string) error {
	d, err := json.Marshal(p)
	if err != nil {
		return err
	}
	_, err = http.Post(url, "application/json", bytes.NewBuffer(d))
	if err != nil {
		return err
	}
	return nil
}

//实例化包级别registry变量
var reg = registry{
	registrations: make([]Registration, 0),
	mutex:         new(sync.RWMutex),
}

type RegistryService struct {
}

func (s RegistryService) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	log.Println("Request received")
	switch r.Method {
	case http.MethodPost:
		dec := json.NewDecoder(r.Body)
		var r Registration
		//从json解码
		err := dec.Decode(&r)
		if err != nil {
			log.Println(err)
			w.WriteHeader(http.StatusBadRequest)
			return
		}
		log.Printf("Adding service: %v with URL:%s\n", r.ServiceName,
			r.ServiceURL)
		//从解码结果注册服务
		err = reg.add(r)
		if err != nil {
			log.Println(err)
			w.WriteHeader(http.StatusBadRequest)
			return
		}
	case http.MethodDelete:
		//读取url
		payload, err := ioutil.ReadAll(r.Body)
		if err != nil {
			log.Println(err)
			w.WriteHeader(http.StatusInternalServerError)
			return
		}
		url := string(payload)
		log.Printf("Removing service at URL: %s", url)
		//根据url删除
		if err := reg.remove(url); err != nil {
			log.Println(err)
			w.WriteHeader(http.StatusInternalServerError)
			return
		}
	default: //其他http请求不允许
		w.WriteHeader(http.StatusMethodNotAllowed)
		return
	}
}

var once sync.Once

// 开启心跳一次
func SetupRegistryService() {
	once.Do(func() {
		go reg.heartbeat(3 * time.Second)
	})
}

//心跳检查(这里为什么一定是指针，可能和单例模式有关)
func (r *registry) heartbeat(freq time.Duration) {
	for {
		var wg sync.WaitGroup
		for _, reg := range r.registrations {
			wg.Add(1)
			go func(reg Registration) {
				defer wg.Done()
				successInFirst := true
				for attemps := 0; attemps < 3; attemps++ {
					res, err := http.Get(reg.HeartbeatURL)
					//如果失败
					if err != nil {
						log.Println(err)
					} else if res.StatusCode == http.StatusOK {
						log.Printf("Heartbeat check passed for %v", reg.ServiceName)
						if !successInFirst { //区分之前是否失败过
							r.add(reg)
						}
						break
					}
					//会再尝试的失败后操作
					log.Printf("Heartbeat check failed for %v", reg.ServiceName)
					if successInFirst {
						successInFirst = false
						r.remove(reg.ServiceURL)
					}
					//间隔1s
					time.Sleep(1 * time.Second)
				}
			}(reg)
			wg.Wait()
			//心跳检测的间隔
			time.Sleep(freq)
		}
	}
}
